package com.xzr.demo.mqserver.datacenter;

import com.xzr.demo.common.MqException;
import com.xzr.demo.mqserver.core.Binding;
import com.xzr.demo.mqserver.core.Exchange;
import com.xzr.demo.mqserver.core.MSGQueue;
import com.xzr.demo.mqserver.core.Message;


import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 管理内存中所有的数据
 * 该类后续提供的一些方法，可能会在多线程环境下使用，因此要注意线程安全问题
 *
 * @Author：徐志荣
 * @Date：2023/8/11 下午 6:27
 */
public class MemoryDataCenter {
    //交换机	：key是name，value 是 Exchange 对象
    //ConcurrentHashMap是Java中的一个线程安全的哈希表实现
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //队列	：key是name，value 是 MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //绑定	：key是exchangeName，value 是一个HashMap  【key是queueName，value是Binding对象】
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //消息	：key 是messageId，value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //队列和消息之间的关联： 使用嵌套的 HashMap，key是queueName，value是 LinkedList 【每个元素是个message对象】
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //“未被确认”的消息		： 使用嵌套的HashMap，key是queueName，value是 HashMap【key 是messageId，value 是 Message 对象】
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageMapWaitAck = new ConcurrentHashMap<>();


    /**
     * 插入交换机
     *
     * @param exchange
     * @return
     */
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[ManagerDataCenter] 新交换机添加成功！exchangeName = "+exchange.getName());

    }
    /**
    * 获取交换机
    * @param exchangeName
    * @return
    */
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    /**
    * 删除交换机
    * @param exchangeName
    * @return
    */
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[ManagerDataCenter] 删除交换机成功！exchangeName = "+exchangeName);

    }

    /**
     * 插入队列
     *
     * @param
     * @return
     */
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[ManagerDataCenter] 队列添加成功！exchangeName = "+queue.getName());

    }

    /**
    * 删除队列
    * @param queueName
    * @return
    */
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[ManagerDataCenter] 队列删除成功！exchangeName = "+queueName);

    }

    /**
    * 获取队列
    * @param queueName
    * @return
    */
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    /**
     * 插入绑定
     *
     * @param binding
     * @return
     */
    public void insertBinding(Binding binding) throws MqException {
        //先使用 exchangeName 是否存在， 不存在就创建一个
       /* ConcurrentHashMap<String, Binding> bindingMap = this.bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            bindingMap = new ConcurrentHashMap<>();
            bindingsMap.put(binding.getExchangeName(), bindingMap);
        }*/
        //与上述的代码逻辑等价
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        //针对 bindingMap加锁
        synchronized (bindingMap) {
            //在根据 queueName  查一下， 如果已经存在，就抛出异常，不存在才能插入
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在 exchangeName = " + binding.getExchangeName() +
                        ",queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[ManagerDataCenter] 新的绑定添加成功！exchangeName = "+binding.getExchangeName()
                +"queueName = "+binding.getQueueName());


    }

    /**
     * 获取绑定
     * 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    * @param exchangeName
    * @param queueName
    * @return
    */
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);

    }

    /**
    * 获取绑定
    * 2. 根据exchangeName 获取到所有的 Binding
    * @param exchangeName
    * @return
    */
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    /**
    * 删除绑定
    * @param binding
    * @return
    */
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //该交换机没有绑定任何队列，报错
            throw new MqException("[ManagerDataCenter] 绑定不存在， exchangeName = "
                    + binding.getExchangeName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[ManagerDataCenter] 绑定删除成功！exchangeName = "+binding.getExchangeName()
                +"queueName = "+binding.getQueueName());
    }

    /**
    * 添加消息
    * @param message
    * @return
    */
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(),message);
        System.out.println("[ManagerDataCenter] 新消息添加成功！messageId = "+message.getMessageId());

    }

    /**
    * 根据id 查询消息
    * @param messageId
    * @return
    */
    public Message getMessage(String  messageId) {
        return messageMap.get(messageId);
    }

    /**
    * 根据id 删除消息
    * @param messageId
    * @return
    */
    public void removeMessage(String  messageId) {
        messageMap.remove(messageId);
        System.out.println("[ManagerDataCenter] 消息被移除！messageId = "+messageId);

    }

    /**
     * 把一个新的消息，放进对应的队列数据结构中
     * @param queue 表示要把消息写入到队列
     * @param message 要写的消息
     * @return
     */
    public void sendMessage(MSGQueue queue, Message message) {
         //先根据队列的名字，找到该队列的消息链表
        /*LinkedList<Message> messages = queueMessageMap.get(queue.getName());
        if(messages == null){
            messages = new LinkedList<>();
            queueMessageMap.put(queue.getName(),messages);
        }*/

        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());

        //在把数据加到message里面
        synchronized (messages){
            messages.add(message);

        }
        //在这里把消息也往消息中心中插入一下,假设如果message 已经在消息中心存在，重复插入也没关系
        //主要就是相同 messageId ， 对应的message 的内容一定是一样的（服务器代码不会对message 内容进行修改的）
        addMessage(message);
        System.out.println("[ManagerDataCenter] 消息被投递到队列中！messageId = "+message.getMessageId());

    }

    /**
    * 从队列中取消息
    * @param queueName
    * @return
    */
    public Message pollMessage(String queueName){
        //根据队列名，查找一下，对应队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return null;
        }
        synchronized (messages){
            //如果没找到，说明队列中没有任何消息
            if(messages.size() == 0){
                return null;
            }
            //链表中有元素，进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[ManagerDataCenter] 消息从队列中取出！messageId = "+currentMessage.getMessageId());
            return currentMessage;

        }

    }

    /**
    * 获取指定队列中 消息的个数
    * @param queueName
    * @return
    */
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    /**
    * 添加未确认的消息
    * @param queueName
    * @param message
    * @return
    */
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageMapWaitAck.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[ManagerDataCenter] 消息进入待确认队列！messageId = "+message.getMessageId());

    }

    /**
    * 删除未被确认的消息
    * @param queueName
    * @param messageId
    * @return
    */
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageMapWaitAck.get(queueName);
        if(messageHashMap == null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[ManagerDataCenter] 消息从待确认队列删除！messageId = "+messageId);
    }

    /**
    * 获取未被确认的消息
    * @param queueName
    * @param messageId
    * @return
    */
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageMapWaitAck.get(queueName);
        if(messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    /**
    * 实现数据从硬盘上恢复
    *
    * @return
    */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0. 清空之前所有的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();

        for (Exchange exchange:exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }

        //2. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue:queues) {
            queueMap.put(queue.getName(), queue);
        }

        //3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding :bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }

        //4. 恢复所有的消息数据
        //遍历所有的队列，针具每个队列的名字，获取所有的消息
        for(MSGQueue queue:queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for(Message message:messages){
                messageMap.put(message.getMessageId(),message);
            }
        }



    }

    }

